package cn.linjianhui.flink.sample.web.controller;

import cn.linjianhui.flink.sample.web.model.*;
import com.alibaba.fastjson.JSON;
import cn.linjianhui.flink.sample.web.dao.mysql.WareDAO;
import cn.linjianhui.flink.sample.web.dao.mysql2.AllOrderAmountDAO;
import cn.linjianhui.flink.sample.web.dao.oracle.WareBrandDAO;
import cn.linjianhui.flink.sample.web.model.*;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import redis.clients.jedis.Jedis;

import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@RestController
@RequestMapping("/data")
public class DataController {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private WareDAO wareDAO;

    @Autowired
    private WareBrandDAO wareBrandDAO;

    @Autowired
    private AllOrderAmountDAO allOrderAmountDAO;

    @GetMapping("getPayOrder")
    public DataModel getPayOrder() {
        String key = "flink-sample-pay_order";
        String wareCategoryKey = "flink-sample-ware_classify";

        DataModel data = new DataModel(BigDecimal.ZERO, 0L, 0L);

        if (redisTemplate.hasKey(key)) {
            Map<Object, Object> cursor = redisTemplate.opsForHash().entries(key);
            for (Object left : cursor.keySet()) {
                OrderStats item = JSON.parseObject(String.valueOf(cursor.get(left)), OrderStats.class);
                data.addPayNumber(item.getTotalPayOrder());
                data.addPayAmount(new BigDecimal(item.getTotalPayAmount()));
            }
        }

        if (redisTemplate.hasKey(wareCategoryKey)) {
            Map<Object, Object> cursor = redisTemplate.opsForHash().entries(wareCategoryKey);
            for (Object left : cursor.keySet()) {
                WareCategoryStats item = JSON.parseObject(String.valueOf(cursor.get(left)), WareCategoryStats.class);
                data.addQty(item.getQty());
            }
        }

        return data;
    }

    @GetMapping("getTopNProduct")
    public List<WareResponse> getTopNProduct(String virtualCategory, int topN) {
        String key = String.format("flink-sample-sales_%s", virtualCategory);
        List<WareResponse> result = new ArrayList<>();
        if (redisTemplate.hasKey(key)) {
            Set<String> rank = redisTemplate.opsForZSet().reverseRange(key, 0, topN - 1);
            String[] wareSkuCodes = rank.toArray(new String[0]);
            List<WareDO> wares = wareDAO.listWare(Arrays.asList(wareSkuCodes));
            for (String code : wareSkuCodes) {
                Optional<WareDO> single = wares.stream().filter(n -> code.equals(n.getWareCode())).findFirst();
                if (single.isPresent()) {
                    WareResponse item = new WareResponse();
                    BeanUtils.copyProperties(single.get(), item);
                    double qty = redisTemplate.opsForZSet().score(key, code);
                    item.setQty((long) qty);
                    result.add(item);
                }
            }
        }
        return result;
    }

    @GetMapping("totalPayCustomer")
    public Long totalPayCustomer() {
        String key = "flink-sample-customer";

        if (!redisTemplate.hasKey(key)) return 0L;
        Jedis jedis = (Jedis)redisTemplate.getConnectionFactory().getConnection().getNativeConnection();
//        jedis.evalsha("", 1, )
        long size = redisTemplate.opsForHyperLogLog().size(key);

        return size;
    }

    @GetMapping("getPayAmountByArea")
    public List<AreaPayStats> listPayAmountByArea() {
        String key = "flink-sample-area_pay";

        List<String> liveProvinceCodes = new ArrayList<>();
        List<AreaPayStats> areas = new ArrayList<>();
        if (redisTemplate.hasKey(key)) {
            Map<Object, Object> cursor = redisTemplate.opsForHash().entries(key);
            for (Object left : cursor.keySet()) {
                AreaPayStats item = JSON.parseObject(String.valueOf(cursor.get(left)), AreaPayStats.class);
                item.setProvince(ProvinceEnum.toName(item.getProvinceCode()));
                areas.add(item);
                liveProvinceCodes.add(item.getProvinceCode());
            }
        }
        for (ProvinceEnum province : ProvinceEnum.values()) {
            if (!liveProvinceCodes.contains(province.getValue())) {
                areas.add(new AreaPayStats(BigDecimal.ZERO, province.name(), province.getValue()));
            }
        }

        return areas;
    }

    @GetMapping("listSalesAmountByWareCategory")
    public List<WareCategoryStats> listSalesAmountByWareCategory() {
        String key = "flink-sample-ware_classify";

        List<WareCategoryStats> stats = new ArrayList<>();

        if (redisTemplate.hasKey(key)) {
            Map<Object, Object> cursor = redisTemplate.opsForHash().entries(key);
            for (Object left : cursor.keySet()) {
                WareCategoryStats item = JSON.parseObject(String.valueOf(cursor.get(left)), WareCategoryStats.class);
                stats.add(item);
            }
            stats = stats.stream().sorted((prev, next) -> {
                BigDecimal x = prev.getSalesAmount();
                BigDecimal y = next.getSalesAmount();
                return y.compareTo(x);
            }).collect(Collectors.toList());
            this.setCacheExpire(key);
        }

        return stats;
    }

    @GetMapping("getTopNSalesByBrand")
    public List<BrandSalesResult> getTopNSalesByBrand(int topN) {
        String key = "flink-sample-sales_brand";
        List<BrandSalesResult> result = new ArrayList<>();
        if (redisTemplate.hasKey(key)) {
            Set<String> rank = redisTemplate.opsForZSet().reverseRange(key, 0, topN - 1);
            String[] brandCodes = rank.toArray(new String[0]);
            List<WareBrandDO> brands = wareBrandDAO.listWareBrand(Arrays.asList(brandCodes));
            for (String code : brandCodes) {
                Optional<WareBrandDO> single = brands.stream().filter(n -> code.equals(n.getBrandCode())).findFirst();
                if (single.isPresent()) {
                    BrandSalesResult item = new BrandSalesResult();
                    BeanUtils.copyProperties(single.get(), item);
                    double amount = redisTemplate.opsForZSet().score(key, code);
                    item.setSalesAmount(amount);
                    result.add(item);
                }
            }
        }
        return result;
    }

    @GetMapping("getOrderAmount")
    public OrderAmountResponse getOrderAmount() {
        LocalDate today = LocalDate.now();
        LocalDate yesterday = LocalDate.now().plusDays(-1);

        OrderAmountResponse response = new OrderAmountResponse();
        response.setToday(allOrderAmountDAO.listOrderAmount(today.toString(), today.toString()));
        response.setYesterday(allOrderAmountDAO.listOrderAmount(yesterday.toString(), yesterday.toString()));

        return response;
    }

    private void setCacheExpire(String key) {
        if (redisTemplate.getExpire(key) == -1) {
            long todayTimestamp = LocalDateTime.now().toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
            long tomorrowTimestamp = LocalDate.now().plusDays(1).atStartOfDay(ZoneOffset.ofHours(8)).toInstant().toEpochMilli();
            redisTemplate.expire(key, (tomorrowTimestamp - todayTimestamp) / 1000, TimeUnit.SECONDS);
        }
    }


    @GetMapping("getPayAmountByChannel")
    public List<SaleChannelStas> listPayAmountByChannel() {
        String key = "flink-sample-sale_channel";
        List<SaleChannelStas> result = new ArrayList<>();
        if (redisTemplate.hasKey(key)) {
            Set<String> keys = redisTemplate.opsForZSet().range(key, 0, 3);
            String[] channels = keys.toArray(new String[0]);
            for (String code : channels) {
                double amount = redisTemplate.opsForZSet().score(key, code);
                SaleChannelStas saleChannelStas = new SaleChannelStas();
                saleChannelStas.setSaleChannel(code);
                saleChannelStas.setSalesAmount(new BigDecimal(amount));
                result.add(saleChannelStas);
            }
        }
        String[] channels = {"新零售", "第三方"};
        for (String saleChannel : channels) {
            if (!result.stream().anyMatch(t -> t.getSaleChannel().equals(saleChannel))) {
                SaleChannelStas saleChannelStas = new SaleChannelStas();
                saleChannelStas.setSaleChannel(saleChannel);
                saleChannelStas.setSalesAmount(BigDecimal.ZERO);
                result.add(saleChannelStas);
            }
        }
        return result.stream().sorted(Comparator.comparing(SaleChannelStas::getSalesAmount).reversed()).collect(Collectors.toList());
    }
}
